package com.codeduck.kafka.handler;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * @auther Mahone Yan
 * @date 2023/5/7
 */
@Component("ServerLogMessageHandler")
public class ServerLogMessageHandler extends IMessageHandler{
    private static final Logger LOGGER = LoggerFactory.getLogger(ServerLogMessageHandler.class);

    @Override
    public void batchConsume(ConsumerRecords<Object, Object> records) {
        List<Object> messages = new ArrayList<>(records.count());
        for (ConsumerRecord<Object, Object> record : records) {
            messages.add(record.value());
        }
        LOGGER.info("推送日志信息至ElasticSearch，批量数据大小[{}], 日志信息[{}]", messages.size(), JSON.toJSONString(messages));
    }
}
